package us.luosl.hadoop.hive.udaf;

import org.apache.hadoop.hive.ql.exec.UDAF;
import org.apache.hadoop.hive.ql.exec.UDAFEvaluator;

/**
 * 老版本的 udaf 实现
 *

 add jar /mr-1.0-SNAPSHOT.jar;
 CREATE TEMPORARY FUNCTION my_avg AS 'us.luosl.hadoop.hive.udaf.UDAFAvg';
 select my_avg(yka055) from kc21k1;

 * Created by luosl on 2018/9/5.
 */
public class UDAFAvg extends UDAF{

    private static class AVGBuffer{
        private double sum = 0D;
        private long count = 0L;
    }

    public static class AvgEvaluator implements UDAFEvaluator {

        private AVGBuffer buff;

        public AvgEvaluator() {
            init();
        }

        @Override
        public void init() {
            buff = new AVGBuffer();
        }

        /**
         * iterate 接收传入的参数, 并进行内部轮转, 返回类型为 boolean
         * @param value value
         * @return boolean
         */
        public boolean iterate(Double value) {
            if (value != null) {
                buff.sum += value;
            }
            buff.count += 1;
            return true;
        }

        /**
         * 在计算平均值时我们不能进行 combiner 所以这里返回本身
         * @return AVGBuffer
         */
        public AVGBuffer terminatePartial () {
            return buff;
        }

        /**
         * 接收 terminatePartial 的返回结果, 进行数据 merge 操作
         * @param buff buff
         * @return boolean
         */
        public boolean merge(AVGBuffer buff) {
            this.buff.count += buff.count;
            this.buff.sum += buff.sum;
            return true;
        }

        /**
         * 返回最终的聚集函数结果
         * @return Double
         */
        public Double terminate () {
            if (buff.count == 0) {
                return null;
            } else {
                return buff.sum / buff.count;
            }
        }

    }

}
